/**
 * @author: yyq
 * @date: 2025年3月14日 13:05:36
 * @description: 
 */
package com.sheepone.monitor.runner;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import com.sheepone.monitor.config.MonitorConfig;
import com.sheepone.monitor.core.AbstractSheeponeChannelInitializer;
import com.sheepone.monitor.runner.listener.MonitorRunnerListener;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

/**
 * @author: yyq
 * @date: 2025年3月14日 13:05:36
 * @description:
 */
public abstract class AbstractMonitorRunner<C extends MonitorConfig, I extends AbstractSheeponeChannelInitializer>
        implements MonitorRunnerListener {
    protected C config;
    protected AtomicBoolean running = new AtomicBoolean(false);
    private List<MonitorRunnerListener> listeners = new ArrayList<>();

    public AbstractMonitorRunner(C config) {
        this.config = config;
    }

    protected abstract void shutdown();

    protected abstract ChannelFuture start(I channelInitializer) throws Exception;

    protected void shutdownGracefully() {
        if (!running.compareAndSet(true, false)) {
            return;
        }
        shutdown();
    }

    public void addListener(MonitorRunnerListener listener) {
        if (listener == null | listeners.contains(listener)) {
            return;
        }
        listeners.add(listener);
    }

    public void beforeStart() {
    }

    public void startSuccessfully(Channel channel) {
        for (MonitorRunnerListener listener : listeners) {
            listener.startSuccessfully(channel);
        }
    }

    public void startFailed(Channel channel, Throwable cause) {
        for (MonitorRunnerListener listener : listeners) {
            listener.startFailed(channel, cause);
        }
    }

    public void stopSuccessfully(Channel channel) {
        for (MonitorRunnerListener listener : listeners) {
            listener.stopSuccessfully(channel);
        }
    }

    public void stopFailed(Channel channel, Throwable cause) {
        for (MonitorRunnerListener listener : listeners) {
            listener.stopFailed(channel, cause);
        }
    }

    private ChannelFuture registerChannelFutureListeners(ChannelFuture cf) {
        Channel channel = cf.channel();
        // ! 监听 channel bind or connect 是否成功
        cf.addListener((future) -> {
            if (future.isSuccess()) {
                running.set(true);
                startSuccessfully(channel);
            } else {
                startFailed(channel, future.cause());
            }
        });
        // ! 监听 channel 关闭
        ChannelFuture closeFuture = channel.closeFuture();
        closeFuture.addListener((future) -> {
            if (future.isSuccess()) {
                stopFailed(channel, future.cause());
            }
            shutdownGracefully();
            stopSuccessfully(channel);
        });
        return closeFuture;
    }

    public void run(I channelInitializer) {
        beforeStart();
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            shutdownGracefully();
        }));
        ChannelFuture cf = null;
        try {
            cf = start(channelInitializer);
            ChannelFuture closeFuture = registerChannelFutureListeners(cf);
            // ! 阻塞当前线程
            closeFuture.sync();
        } catch (Exception e) {
            Channel channel = cf.channel();
            startFailed(channel, e);
            shutdownGracefully();
            stopSuccessfully(channel);
        }
    }
}
